package com.liser.southernstar.utils;

import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import io.reactivex.schedulers.Schedulers;

/**
 * 高性能流工具类
 * <ul>
 *      主要功能速览
 *     <li>{@linkplain #close(Closeable)} ：关闭流，忽略关闭异常</li>
 *     <li>{@linkplain #tryCopy(InputStream, byte[])}：读取直到缓冲区满或发生IO错误</li>
 *     <li>{@linkplain #copy(InputStream, byte[])}：读取直到缓冲区满或到达流末尾，返回有效数量</li>
 *     <li>{@linkplain #copy(byte[], OutputStream)}：输出并立即刷新缓冲区</li>
 *     <li>{@linkplain #copy(InputStream, OutputStream, int)}：单线程IO拷贝</li>
 *     <li>{@linkplain #fastCopy(InputStream, OutputStream, int)}：双线程IO拷贝</li>
 * </ul>
 */
public class StreamUtils {

    /**
     * 关闭流，忽略关闭期间的异常
     * @param closeable 流
     */
    public static void close(Closeable closeable){
        if(closeable != null){
            try {
                closeable.close();
            }catch (Exception ex){
            }
        }
    }

    /**
     * 读取缓冲区，直到填满为止
     * @param in 输入流
     * @param out 缓冲
     * @return out参数
     * @throws IOException 如果数据未填满缓冲区但已读取到流末尾。或其它原因
     */
    public static byte[] tryCopy(InputStream in, byte[] out) throws IOException {
        for(int read = 0, ret, readCount = out.length;;){
            ret = in.read(out, read, readCount - read);
            if(ret == -1){
                throw new IOException("流已到达末尾，缓冲区数据已截断！");
            }
            if((read+=ret) == readCount){
                break;
            }
        }
        return out;
    }

    /**
     * 读取缓冲区，直到填满为止或已到达流末尾
     * @param in 输入流
     * @param out 缓冲区
     * @return 实际读取的byte数，返回0时已到达流末尾
     */
    public static int copy(InputStream in, byte[] out) throws IOException {
        int read = 0;
        for(int ret, readCount = out.length;;){
            ret = in.read(out, read, readCount - read);
            if(ret == -1 || (read+=ret) == readCount){
                break;
            }
        }
        return read;
    }

    /**
     * 将缓冲区的数据写入流并立即调用flush
     * @param in 缓冲
     * @param out 输出流
     */
    public static void copy(byte[] in, OutputStream out) throws IOException {
        copy(in, 0, in.length, out);
    }

    /**
     * 将缓冲区的指定byte数量写入流并立即调用flush
     * @param in 缓冲
     * @param count 写入的数量
     * @param out 输出流
     */
    public static void copy(byte[] in, int count, OutputStream out) throws IOException {
        copy(in, 0, count, out);
    }

    /**
     * 将缓冲区的指定byte段写入流并立即调用flush
     * @param in 缓冲
     * @param offset 起始位置
     * @param count 写入的数量
     * @param out 输出流
     */
    public static void copy(byte[] in, int offset, int count, OutputStream out) throws IOException {
        out.write(in, offset, count);
        out.flush();
    }

    /**
     * 流复制，默认缓冲区为8kb
     * @param in 输入流
     * @param out 输出流
     */
    public static void copy(InputStream in, OutputStream out) throws IOException {
        copy(in, out, 8192);
    }

    /**
     * 流复制
     * @param in 输入流
     * @param out 输出流
     * @param bufferSize 缓冲区大小
     */
    public static void copy(InputStream in, OutputStream out, int bufferSize) throws IOException {
        byte[] buffer = new byte[bufferSize];
        for(int read;;){
            read = copy(in, buffer);
            if(read == 0 ){
                break;
            }
            copy(buffer, read, out);
        }
    }

    /**
     * 双线程流复制，适用于双方速率低下的IO拷贝<br/>
     * 使用当前线程读，开辟新线程写。默认缓冲区为8kb
     * @param in 输入流
     * @param out 输出流
     */
    public static void fastCopy(InputStream in, OutputStream out) throws IOException {
        fastCopy(in, out, 8192);
    }

    /**
     * 双线程流复制，适用于双方速率低下的IO拷贝<br/>
     * 使用当前线程读，开辟新线程写
     * @param in 输入流
     * @param out 输出流
     * @param bufferSize 缓冲区大小
     */
    public static void fastCopy(InputStream in, OutputStream out, int bufferSize) throws IOException {
        // 读缓存
        byte[] readBuffer;
        byte[] bufferA = new byte[bufferSize];
        byte[] bufferB = new byte[bufferSize];
        // 写缓存
        final CallData<byte[]> writeBuffer = new CallData<>();
        final CallData<Integer> writeLength = new CallData<>();
        // 错误信息
        final CallData<Exception> exception = new CallData<>();
        // 同步信号
        AutoResetEvent readSignal = new AutoResetEvent(true);
        AutoResetEvent writeSignal = new AutoResetEvent(false);
        CallData<Boolean> exit = new CallData<>(false);
        //写线程
        new RxJava2Task<Void>((emitter) -> {
            try{
                for(;;){
                    // 等待读线程
                    try{writeSignal.waitOne();}catch (Exception e){}
                    if(exit.getValue()){
                        return;
                    }
                    // 写数据
                    copy(writeBuffer.getValue(), writeLength.getValue(), out);
                    // 唤醒读线程
                    readSignal.set();
                }
            }catch (Exception ex){
                exception.setValue(ex);
            }finally {
                exit.setValue(true);
                readSignal.set();
            }
        }, Schedulers.io()).subscribeOnce();
        // 读线程
        try{
            for(boolean isBufferA = false;;){
                if(isBufferA){
                    readBuffer = bufferB;
                    isBufferA = false;
                }else {
                    readBuffer = bufferA;
                    isBufferA = true;
                }
                // 缓存远程数据并等待写线程
                int read = copy(in, readBuffer);
                try{readSignal.waitOne();}catch (Exception e){}
                if(read == 0 || exit.getValue()){
                    return;
                }
                // 设置写缓存并唤醒写线程
                writeBuffer.setValue(readBuffer);
                writeLength.setValue(read);
                writeSignal.set();
            }
        }catch (Exception ex){
            exception.setValue(ex);
        }finally {
            exit.setValue(true);
            writeSignal.set();
        }
        // 异常处理
        Exception ex = exception.getValue();
        if(ex == null){
            return;
        }
        throw ex.getClass() == IOException.class ? (IOException)ex : new IOException(ex);
    }
}